from kafka import KafkaConsumer
import json
import time

# 创建消费者实例
consumer = KafkaConsumer(
    bootstrap_servers=['node1:9092', 'node2:9092', 'node3:9092', 'node4:9092'],  # 替换为你的Kafka服务器地址
    auto_offset_reset='earliest',  # 从最早的消息开始读取（'latest'表示仅读取新消息）
    enable_auto_commit=True,  # 自动提交偏移量
    group_id='test_group',  # 确保所有消费者实例使用相同的消费者组ID
    value_deserializer=lambda x: json.loads(x.decode('utf-8')) if x else None  # 反序列化消息值
)

# 订阅主题
consumer.subscribe(['test'])  # 替换为你的主题名称

print("Starting to poll for messages...")

try:
    while True:
        # 轮询消息，设置超时时间为1秒（1000毫秒）
        msg_pack = consumer.poll(timeout_ms=1000, max_records=5)  # 一次最多拉取5条消息

        # 遍历每个分区的消息包
        for tp, messages in msg_pack.items():
            for message in messages:
                print(f"Received message: {message.value} from topic {message.topic} at offset {message.offset}")

                # 处理消息逻辑
                # ...

                # 提交偏移量，确保消息被正确处理
                consumer.commit()
                print("Offset committed.")

except KeyboardInterrupt:
    print("Interrupted by user.")
except Exception as e:
    print(f"An error occurred: {e}")
finally:
    # 关闭消费者连接
    consumer.close()
    print("Consumer closed.")